package com.zuora.interview.topn.processor.step3;

import com.google.common.base.Splitter;
import com.google.common.collect.Maps;
import com.zuora.interview.topn.bean.Constants;
import com.zuora.interview.topn.bean.ThreePagePathResult;
import com.zuora.interview.topn.util.Clock;
import org.apache.commons.io.FileUtils;
import org.apache.commons.io.LineIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * Created by chunyang.gao on 2016/8/23.
 */
public class PathPartitioner {

    private static final Logger console = LoggerFactory.getLogger("console");

    Splitter commaSplitter = Splitter.on(",");

    public void work() throws IOException {
        Clock clock = Clock.start();
        File d = new File(Constants.logPath+"step2");
        final int topn = 5;
        final ConcurrentMap<String,ThreePagePathResult> topNContainer = Maps.newConcurrentMap();
        final CountDownLatch listener = new CountDownLatch(d.listFiles().length);
        for(final File f:d.listFiles()){
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        SortedSet<ThreePagePathResult> singleTopN= new PathPartitioner().test(f,topn);
                        for(ThreePagePathResult tpp:singleTopN){
                            ThreePagePathResult ih = new ThreePagePathResult(tpp.getPath());

                            topNContainer.putIfAbsent(tpp.getPath(),ih);
                            ih.incr(tpp.getCount());
                        }
                    } catch (IOException e) {
                        e.printStackTrace();
                    }finally {
                        listener.countDown();
                    }
                }
            }).start();
        }

        try {
            listener.await(30, TimeUnit.SECONDS);
            TreeSet<ThreePagePathResult> result = new TreeSet<ThreePagePathResult>();
            result.addAll(topNContainer.values());
            ThreePagePathResult breaker = result.first();
            int index = 0;
            int nthValue = 0;

            for(ThreePagePathResult tpp:result){
                index++;
                if(index==topn){
                    nthValue = tpp.getCount();
                    breaker = tpp;
                }
                if(index>topn){
                    if(nthValue == tpp.getCount()){
                        breaker = tpp;
                    }
                    if(nthValue > tpp.getCount()){
                        break;
                    }
                }
            }

            console.warn("step3 in {} ms as {}", clock.checkTime(), result.headSet(breaker, true));
        } catch (InterruptedException e) {
            console.warn("step3 in " + clock.checkTime() + " ms as error", e);
        }
    }

    private SortedSet<ThreePagePathResult> test(File f,int topn) throws IOException {
        LineIterator it = FileUtils.lineIterator(f);
        Map<String,Integer> countBy3PP = Maps.newHashMap();
        try {
            while (it.hasNext()) {
                String line = it.nextLine();
                List<String> parts = commaSplitter.splitToList(line);
                String threePagePath = parts.get(0);
                Integer count = countBy3PP.get(threePagePath);
                if(null==count){
                    count=0;
                }
                countBy3PP.put(threePagePath, count + 1);
            }
        } finally {
            LineIterator.closeQuietly(it);
        }
        TreeSet<ThreePagePathResult> result = new TreeSet<ThreePagePathResult>();
        for(String threePagePath:countBy3PP.keySet()){
            result.add(new ThreePagePathResult(threePagePath,countBy3PP.get(threePagePath)));
        }
        ThreePagePathResult breaker = result.first();
        if(result.size()<=topn){
            return result;
        }else{
            int index = 0;
            int nthValue = 0;

            for(ThreePagePathResult tpp:result){
                index++;
                if(index==topn){
                    nthValue = tpp.getCount();
                    breaker = tpp;
                }
                if(index>topn){
                    if(nthValue == tpp.getCount()){
                        breaker = tpp;
                    }
                    if(nthValue > tpp.getCount()){
                        break;
                    }
                }
            }
        }
        return result.headSet(breaker,true);
    }

}
